﻿using log4net;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using RabbitMQ.Client.Exceptions;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ConsumerMQ.Models
{
    /// <summary>
    /// 監聽基類 (审计消费者)
    /// 創建子類，繼承它，并實現Process方法（處理消息的方法）
    /// </summary>
    public class AuditRabbitListener : BackgroundService
    {
        private readonly IConnection connection;
        private readonly IModel channel;
        private readonly string ExchangeName;
        private readonly string QueueName;
        private readonly List<int> retryTime;
        private ILog log;

        public AuditRabbitListener(IOptions<RabbitMQSetting> options)
        {
            try
            {
                //创建连接工厂对象
                var factory = new ConnectionFactory()
                {
                    // 这是我这边的配置,自己改成自己用就好
                    HostName = options.Value.RabbitHost,
                    UserName = options.Value.RabbitUserName,
                    Password = options.Value.RabbitPassword,
                    Port = options.Value.RabbitPort,
                };
                this.connection = factory.CreateConnection();
                this.channel = connection.CreateModel();

                this.ExchangeName = "ExchangeName_GLZ";
                this.QueueName = "QueueName_GLZ";

                this.retryTime = new List<int>
                {
                    1 * 1000,
                    10 * 1000,
                    30 * 1000
                };

                this.log = LogManager.GetLogger(Startup.repository.Name, typeof(RabbitListener));
                log.Info("init AuditRabbitListener");
            }
            catch (Exception ex)
            {
                log.Error($"AuditRabbitListener init error,ex:{ex.Message}");
            }
        }

        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            throw new NotImplementedException();
        }

        public override Task StopAsync(CancellationToken cancellationToken)
        {
            this.connection.Close();
            return Task.CompletedTask;
        }

        public override Task StartAsync(CancellationToken cancellationToken)
        {
            Register();
            return Task.CompletedTask;
        }

        protected string RouteKey;

        // 处理消息的方法
        public virtual bool Process(string message)
        {
            throw new NotImplementedException();
        }

        // 注册消费者监听在这里
        public void Register()
        {
            log.Info($"AuditRabbitListener register,routeKey:{RouteKey}");

            channel.BasicQos(0, 1, false);  //设置预取消息

            channel.ExchangeDeclare(exchange: ExchangeName, type: "topic");

            //声明一个队列
            channel.QueueDeclare(queue: QueueName,//消息队列名称
                  durable: true,//是否缓存
                exclusive: false,
               autoDelete: false,
                arguments: null);

            channel.QueueBind(queue: QueueName, //消息队列名称
                              exchange: ExchangeName,
                              routingKey: RouteKey);


            //设置死信交换机(Dead Letter Exchange)
            var retryDic = new Dictionary<string, object>
            {
                {"x-dead-letter-exchange", "Exchange"},
                {"x-dead-letter-routing-key", "RouteB"}
            };

            channel.ExchangeDeclare("Exchange_Retry", "direct");
            channel.QueueDeclare("QueueB_Retry", true, false, false, retryDic);
            channel.QueueBind("QueueB_Retry", "Exchange_Retry", "RouteB_Retry");


            //创建消费者对象
            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                bool canAck;
                var retryCount = 0;
                if (ea.BasicProperties.Headers != null && ea.BasicProperties.Headers.ContainsKey("retryCount"))
                {
                    retryCount = (int)ea.BasicProperties.Headers["retryCount"];
                    log.Info($"[{DateTime.Now:yyyy-MM-dd HH:mm:ss}]Message:{ea.BasicProperties.MessageId}, {++retryCount} retry started...");
                }

                try
                {
                    var body = ea.Body; //接收到的消息
                    var message = Encoding.UTF8.GetString(body.ToArray());
                    var result = Process(message);
                    if (result)
                    {
                        canAck = true;
                    }
                    else
                    {
                        canAck = false;
                    }
                }
                catch (Exception ex)
                {
                    log.Error($"Error:{ex}");
                    if (CanRetry(retryCount))
                    {
                        SetupRetry(retryCount, "Exchange_Retry", "RouteB_Retry", ea);
                        canAck = true;
                    }
                    else
                    {
                        canAck = false;
                    }
                }

                try
                {
                    if (canAck)
                    {
                        channel.BasicAck(ea.DeliveryTag, false);
                    }
                    else
                    {
                        channel.BasicNack(ea.DeliveryTag, false, false);
                    }
                }
                catch (AlreadyClosedException ex)
                {
                    log.Error($"RabbitMQ is closed！error:{ex}");
                }
            };
            //消费者开启监听
            channel.BasicConsume(queue: QueueName, false, consumer: consumer);
        }

        /// <summary>
        /// 可以重试，则启动重试机制
        /// </summary>
        /// <param name="retryCount"></param>
        /// <param name="retryExchange"></param>
        /// <param name="retryRoute"></param>
        /// <param name="ea"></param>
        private void SetupRetry(int retryCount, string retryExchange, string retryRoute, BasicDeliverEventArgs ea)
        {
            var body = ea.Body;
            var properties = ea.BasicProperties;
            properties.Headers = properties.Headers ?? new Dictionary<string, object>();
            properties.Headers["retryCount"] = retryCount;
            properties.Expiration = retryTime[retryCount].ToString();

            try
            {
                channel.BasicPublish(retryExchange, retryRoute, properties, body);
            }
            catch (AlreadyClosedException ex)
            {
                log.Error($"RabbitMQ is closed！error:{ex}");
            }
        }

        /// <summary>
        /// 发生异常，判断是否可以重试
        /// </summary>
        /// <param name="retryCount"></param>
        /// <returns></returns>
        private bool CanRetry(int retryCount)
        {
            return retryCount <= retryTime.Count - 1;
        }

        public override void Dispose()
        {
            this.connection.Close();
            base.Dispose();
        }

    }
}
